package com.hdu.dwh.dao;

import com.hdu.dwh.enums.HttpStatusEnum;
import com.hdu.dwh.exception.CustomException;
import com.hdu.dwh.pojos.source_access_pojos.SourceAccess;
import com.hdu.dwh.utils.StatementUtils;

import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.catalog.CatalogDescriptor;
import org.springframework.stereotype.Component;

import java.util.Arrays;

@Slf4j
@Component
@RequiredArgsConstructor
@Data
public class FlinkBaseDao {

    private final TableEnvironment tableEnvironment;

    /**
     * 查看 flink里 的 所有 catalog
     */
    public String[] showCatalog() {
        try {
            return tableEnvironment.listCatalogs();
        } catch (Exception e) {
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }

    public boolean checkCatalogExists(String catalogName) {
        try {
            String[] catalogs = showCatalog();
            return Arrays.asList(catalogs).contains(catalogName);
        } catch (Exception e) {
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }

    /**
     * 初始化 flink 的 mysql/postgresql 连接 catalog
     */
    public boolean initFlinkOther(SourceAccess sourceAccess){
        try {
            Configuration configuration = new Configuration();
            // 参数设置
            configuration.setString("default-database", sourceAccess.getDefaultDatabase());
            configuration.setString("username", sourceAccess.getUsername());
            configuration.setString("password", sourceAccess.getPassword());
            configuration.setString("type", "jdbc");
            configuration.setString("base-url",  String.format("jdbc:%s://%s:%s", sourceAccess.getSourceType(), sourceAccess.getHost(), sourceAccess.getPort()));
            // 创建Catalog描述符
            CatalogDescriptor catalogDescriptor = CatalogDescriptor.of(sourceAccess.getConnectionName(), configuration);
            // 创建Catalog
            tableEnvironment.createCatalog(catalogDescriptor.getCatalogName(), catalogDescriptor);
            return true;
        } catch (Exception e) {
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }


    /**
     * 查看 catalog 里 的 所有 数据库
     */
    public String[] showDatabase(String catalogName) {
        try {
            if(!checkCatalogExists(catalogName)){
                throw new CustomException(HttpStatusEnum.CATALOG_NOT_EXISTS);
            }
            tableEnvironment.useCatalog(catalogName);
            return tableEnvironment.listDatabases();
        } catch (Exception e) {
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }


    /**
     * 查看 catalog 的 某数据库里的 所有 数据表
     */
    public String[] showTable(String catalogName, String databaseName){
        try {
            if(!checkDatabaseExists(catalogName, databaseName)){
                throw new CustomException(HttpStatusEnum.DATABASE_NOT_EXISTS);
            }
            // tableEnvironment.useCatalog(catalogName);
            // tableEnvironment.useDatabase(databaseName);
            return tableEnvironment.listTables(catalogName, databaseName);
            // return tableEnvironment.listTables();
        } catch (Exception e) {
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }


    /**
     * 检查 catalog 里 数据库 是否存在
     */
    public boolean checkDatabaseExists(String catalogName, String databaseName) {
        try {
            if(!checkCatalogExists(catalogName)){
                throw new CustomException(HttpStatusEnum.CATALOG_NOT_EXISTS);
            }
            String[] database = showDatabase(catalogName);
            return Arrays.asList(database).contains(databaseName);
        } catch (Exception e) {
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }


    /**
     * 检查 catalog 里 某数据库 的 数据表是否存在
     */
    public boolean checkTableExists(String catalogName, String databaseName, String tableName) {
        try {
            if(!checkDatabaseExists(catalogName, databaseName)){
                throw new CustomException(HttpStatusEnum.DATABASE_NOT_EXISTS);
            }
            String[] tables = showTable(catalogName, databaseName);
            return Arrays.asList(tables).contains(tableName);
        } catch (Exception e) {
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }


    /**
     * 在catalog里创建数据库
     */
    public boolean createDatabase(String catalogName, String databaseName) {
        try{
            if(!checkDatabaseExists(catalogName, databaseName)){
                tableEnvironment.executeSql(String.format("CREATE DATABASE `%s`.`%s`", catalogName, databaseName));
                return true;
            }else{
                return false;
            }
        }catch (Exception e){
            e.printStackTrace();
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }



    /**
     * 在catalog里删除数据库
     * @param databaseName
     */
    public boolean deleteDatabase(String catalogName, String databaseName) {
        databaseName = StatementUtils.quote(databaseName);
        try{
            if(checkDatabaseExists(catalogName, databaseName)){
                log.info("当前数据库: " + tableEnvironment.getCurrentDatabase());
                tableEnvironment.executeSql(String.format("DROP DATABASE %s.%s CASCADE", catalogName, databaseName));
                return true;
            }else{
                return false;
            }
        }catch (Exception e){
            e.printStackTrace();
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
    }

    /**
     * 删除flink中的表
     * @param catalog
     * @param databaseName
     * @param tableName
     */
    public void deleteTable(String catalog, String databaseName, String tableName) {
        log.info("删除flink中的表: " + catalog + "." + databaseName + "." + tableName);
        try{
            if(checkTableExists(catalog, databaseName, tableName)){
                log.info("当前数据库: " + tableEnvironment.getCurrentDatabase());
                tableEnvironment.executeSql(String.format("DROP TABLE %s.%s.%s", StatementUtils.quote(catalog), StatementUtils.quote(databaseName), StatementUtils.quote(tableName)));
            }
        }catch (Exception e){
            e.printStackTrace();
            throw new CustomException(HttpStatusEnum.TABLE_NOT_EXISTS);
        }
    }
}